package transaction;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.nio.charset.Charset;

/**
 * transaction包,事务消息。half message 半消息，还未设置提交或者回滚的消息，为半消息。
 * 生产者设置事务监听器，监听器中对半消息进行处理。决定是提交还是回滚事务。
 * @author itbac
 * @email 1218585258@qq.com
 * @date 2020/12/13
 */
public class Producer {
    public static void main(String[] args) throws Exception {
        //1.创建消息生产者 producer，并指定生产者组名 group1
        TransactionMQProducer producer = new TransactionMQProducer("group9");
        //2.指定Nameserver地址,集群用分号;隔开 。
        // （注意防火墙要开放指定端口,Nameserver端口9876 ,broker端口10911,broker要设置配置文件启动）
        producer.setNamesrvAddr("192.168.0.88:9876");

        //TODO 设置事务监听器
        producer.setTransactionListener(new TransactionListener() {
            //在该方法中执行本地事务。在收到半消息时回调。
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                System.out.println("executeLocalTransaction:" + new String(message.getBody()));
                if (StringUtils.equals("tagA",message.getTags())){
                    //提交事务
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                if (StringUtils.equals("tagB",message.getTags())){
                    //回滚事务
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                if (StringUtils.equals("tagC",message.getTags())){
                    //不知道。一段时间后，服务器会回调checkLocalTransaction（）方法
                    return LocalTransactionState.UNKNOW;
                }
                return LocalTransactionState.UNKNOW;
            }
            //该方法是MQ进行消息事务状态回查。executeLocalTransaction 中的 LocalTransactionState.UNKNOW 消息才回调
            // 一段时间后，半消息还不知道是否要提交。MQ再回查。
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("checkLocalTransaction:" + new String(messageExt.getBody()));
                //再次回查
                //提交事务
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });

        //3.启动producer
        producer.start();

        //4.创建消息对象，指定主题Topic、Tag 和消息体
        /**
         *参数1：topic主题 （大分类）
         *参数2：tag （小分类）
         *参数3：body （消息内容）
         */
        //组装批量消息
        String[] tags = {"tagA", "tagB", "tagC"};
        for (int i = 0; i < 3; i++) {
            Message message = new Message("transaction", tags[i],
                    ("rocketmq 发送事务消息"+tags[i]).getBytes(Charset.forName("UTF-8")));
            //todo 5.发送事务消息
            //对当前producer所有的消息进行事务控制，这里传null
            producer.sendMessageInTransaction(message, null);
        }
        //6.关闭生产者producer
        //为了执行checkLocalTransaction，生产者要等待一会再停止。
        Thread.sleep(50000);
        producer.shutdown();
    }
}
